package groupOne.app.DWD.db;

import groupOne.app.BaseAppSql;
import groupOne.common.Constant;
import groupOne.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderPreProcess_huangyanhui extends BaseAppSql {
    public static void main(String[] args) {
        new DwdTradeOrderPreProcess_huangyanhui().init(
                5000,
                2,
                "DwdTradeOrderPreProcess_huangyanhui"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {

        // 1. 读取ods_db
        readOdsDb(tEnv,"DwdTradeOrderPreProcess_huangyanhui");
        //2. 过滤出order_detail 的数据
        Table orderDetail = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['course_id'] course_id, " +
                "data['course_name'] course_name, " +
                "data['order_id'] order_id, " +
                "data['user_id'] user_id, " +
                "data['origin_amount'] split_origin_amount, " +
                "data['coupon_reduce'] split_coupon_reduce, " +
                "data['final_amount'] split_final_amount, " +
                "data['session_id'] session_id, " +
                "data['create_time'] create_time, " +
                "ts od_ts, " +
                "pt " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and `type`='insert'");

        tEnv.createTemporaryView("order_detail",orderDetail);

        //3. 过滤order_info
        Table orderInfo = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['origin_amount'] origin_amount, " +
                "data['coupon_reduce'] coupon_reduce, " +
                "data['final_amount'] final_amount, " +
                "data['order_status'] order_status, " +
                "data['session_id'] session_id, " +
                "data['province_id'] province_id, " +
                "data['update_time'] update_time, " +
                "`type`, " +
                "`old`, " +
                "ts oi_ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_info' " +
                "and (`type`='insert'" +
                "     or `type`='update'" +
                ")");
        tEnv.createTemporaryView("order_info",orderInfo);

        //4. 两张表join
        Table result = tEnv.sqlQuery("select " +
                " od.id, " +
                " od.order_id, " +
                " oi.user_id, " +
                " oi.order_status, " +
                " course_id, " +
                " course_name, " +
                " split_origin_amount, " +
                " split_coupon_reduce, " +
                " split_final_amount, " +
                " date_format(od.create_time, 'yyyy-MM-dd') date_id, " +
                " od.create_time, " +
                " date_format(oi.update_time, 'yyyy-MM-dd') update_date_id, " +
                " oi.update_time, " +
                " origin_amount, " +
                " coupon_reduce, " +
                " final_amount, " +
                " od.session_id, " +
                " province_id, " +
                " oi.`type`, " +
                " oi.`old`, " +
                " od.od_ts, " +
                " oi.oi_ts, " +
                " current_row_timestamp() row_op_ts" +
                " from order_detail od " +
                " join order_info oi on od.order_id=oi.id ");

        //5. 写入kafka
        tEnv.executeSql("create table dwd_trade_order_pre_process( " +
                " id string, " +
                " order_id string, " +
                " user_id string, " +
                " order_status string, " +
                " course_id string, " +
                " course_name string, " +
                " split_origin_amount string, " +
                " split_coupon_reduce string, " +
                " split_final_amount string, " +
                " create_time string, " +
                " date_id string, " +
                " update_date_id string, " +
                " update_time string, " +
                " origin_amount string, " +
                " coupon_reduce string, " +
                " final_amount string, " +
                " session_id string, " +
                " province_id string, " +
                " `type` string, " +
                " `old` map<string,string>, " +
                " od_ts bigint, " +
                " oi_ts bigint, " +
                " row_op_ts timestamp_ltz(3)" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_PRE_PROCESS)
        );
        result.executeInsert("dwd_trade_order_pre_process");

    }
}
